Flink对接kafka 之kafka source和kafka sink

您所在的位置:网站首页 flink消费kafka机制 Kafka数据还在吗 Flink对接kafka 之kafka source和kafka sink

Flink对接kafka 之kafka source和kafka sink

2024-06-02 22:19| 来源: 网络整理| 查看: 265

Table of Contents

 

导入连接器依赖

Flink之kafka消费者

反序列化器

自定义反序列化器

设置消费起始点

指定offset起始点

kafka消费容错(checkpoint机制)

分区感知

offset提交配置

提取时间戳与生成watermark

Flink之Kafka生产者

序列化器

kafka生产者分区器

kafka生产者容错机制

使用kafka时间戳和flink事件时间

数据丢失

导入连接器依赖

Flink本身没有提供了链接kafka的接口,需要导入相关依赖才可以使用。在flink 1.7之后,flink-connector-kafka自动适配最新版的kafka,但是但是如果使用低版本的kafka,如0.11、0.10、0.9或0.8,则应该使用对应的kafka连接器。

Maven依赖flink版本消费者和生产者类名kafka版本备注flink-connector-kafka-0.8_2.111.0.0FlinkKafkaConsumer08 FlinkKafkaProducer080.8.x内部使用Kafka 的SimpleConsumer API。偏移由Flink提交给ZK。flink-connector-kafka-0.9_2.111.0.0FlinkKafkaConsumer09 FlinkKafkaProducer090.9.x使用新的Consumer API Kafka。flink-connector-kafka-0.10_2.111.2.0FlinkKafkaConsumer010 FlinkKafkaProducer0100.10.x该连接器支持带有时间戳的Kafka消息,以供生产和使用。flink-connector-kafka-0.11_2.111.4.0FlinkKafkaConsumer011 FlinkKafkaProducer0110.11.x由于0.11.x,Kafka不支持Scala 2.10。该连接器支持Kafka事务消息传递,以为生产者提供精准一次性语义。flink-connector-kafka_2.111.7.0FlinkKafka 消费者FlinkKafka 生产者> = 1.0.0这个通用的Kafka连接器会适配Kafka的最新版本。Flink发行版之间可能会更改其使用的客户端版本。从Flink 1.9版本开始,它使用Kafka 2.2.0客户端。Kafka客户端向后兼容0.10.0或更高版本。但是对于Kafka 0.11.x和0.10.x版本,我们建议分别使用专用的flink-connector-kafka-0.11_2.11和flink-connector-kafka-0.10_2.11。

依赖:

org.apache.flink flink-connector-kafka_2.11 1.10.0 导入隐式转换

scala需要导入隐式转换,否则会报错哦。以下代码示例中就不导入了。

import org.apache.flink.api.scala._ Flink之kafka消费者

Flink的kafka消费者类名是 FlinkKafkaConsumer08  (08是kafka版本,如 Kafka 0.9.0.x的对应flink消费者类名为FlinkKafkaConsumer09 )。

创建flink消费者需要传递三个参数:

topic : 一个topic名,或者一个包含多个topic名的list。序列化器 :  用于序列化消息。Properties 对象 : 包含各种配置,如bootstrap.servers,消费者组,0.8及更早版本的kafka还需要一个zk地址用于保存offset。

如:

val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") // only required for Kafka 0.8 properties.setProperty("zookeeper.connect", "localhost:2181") properties.setProperty("group.id", "test") stream = env .addSource(new FlinkKafkaConsumer08[String]("topic", new SimpleStringSchema(), properties)) .print() 反序列化器

flink消费kafka时,需要知道如何将kafka中的二进制的数据转换成java/scala中的对象,flink自带多种反序列化器,除上面例子中的SimpleStringSchema以外,还有:

TypeInformationSerializationSchema(或 TypeInformationKeyValueSerializationSchema) :基于flink的TypeInformation创建schema,如果数据仅在flink之间流转,可以使用该方式,且性能比其他反序列化方式高。

JsonDeserializationSchema (或 JSONKeyValueDeserializationSchema) :可以将kafka中的json数据转换成一个ObjectNode 对象,可以使用该对象的objectNode.get("field").as(Int/String/...)()来访问指定的字段。如果使用的是括号内的反序列化器,则获得的是包含一个k/v 类型的objectNode,不仅包含json中的所有字段,还包含了kafka的元数据信息,如topic、partitions、offset。

AvroDeserializationSchema :用于使用静态的schema读取Avro格式序列化后的数据。可以从Avro 生成的类中推断出schema信息(如AvroDeserializationSchema.forSpecific(...));也可以手动指定schema信息(使用GenericRecords类,由AvroDeserializationSchema.forGeneric(...)生成)。使用Avro序列化需要导入相应依赖,如flink-avro或flink-avro-confluent-registry。

自定义反序列化器

flink提供DeserializationSchema 接口,允许重写里面的T deserialize(byte[] message) 方法来自定义反序列化器。deserialize会对每一条kafka消息进行处理,并返回自定义类型的数据。

以KeyedDeserializationSchema为例,重写deserialize方法,实现返回一个包含topic、key、value的三元组:

public class KafkaDeserializationTopicSchema implements KeyedDeserializationSchema { public KafkaDeserializationTopicSchema(){ } @Override public Tuple3 deserialize(byte[] keyByte, byte[] message, String topic, int partition, long offset) throws IOException { String key = null; String value = null; if (keyByte != null) { key = new String(keyByte, StandardCharsets.UTF_8); } if (message != null) { value = new String(message,StandardCharsets.UTF_8); } return new Tuple3(topic,key, value); } @Override public boolean isEndOfStream(Tuple3 o) { return false; } @Override public TypeInformation getProducedType() { return TypeInformation.of(new TypeHint(){}); } 设置消费起始点

flink可以设置消费kafka的起始点。如:

val env = StreamExecutionEnvironment.getExecutionEnvironment() val myConsumer = new FlinkKafkaConsumer08[String](...) myConsumer.setStartFromEarliest() // 从最早的offset开始消费 myConsumer.setStartFromLatest() // 从最迟的offset开始消费 myConsumer.setStartFromTimestamp(...) // 从指定的时间开始消费 myConsumer.setStartFromGroupOffsets() // 从当前组消费到的offset开始消费(默认的消费策略) val stream = env.addSource(myConsumer)

setStartFromGroupOffsets :从消费者组提交到kafka的offset开始消费,0.8版本的kafka保存在zookeeper里,0.8之后kafka由一个专门用于保存offset的topic。如果没有找到offset,则从 auto.offset.reset 所设置的参数开始消费。

setStartFromEarliest() / setStartFromLatest():从最早/最迟的offset开始消费,如果使用这种模式,则提交的offset会被忽略,不会从提交的offset开始消费。

setStartFromTimestamp(long):从指定的时间戳开始消费。每个分区中,时间戳大于等于这个时间戳的数据都会被消费;如果分区中最新数据的时间戳小于指定的时间戳,则从最新的时间戳开始消费。如果使用这种模式,则提交的offset会被忽略,不会从提交的offset开始消费。

指定offset起始点

还可以手动指定每个分区从某个offset开始消费,如其中"myTopic"是消费的topic,0 1 2 是topic的分区号,23 31 43是offset ,即下一个消费的offset。如果没有指定分区,则会从消费者组消费到的offset开始消费,即回退到setStartFromGroupOffsets 模式。

val specificStartOffsets = new java.util.HashMap[KafkaTopicPartition, java.lang.Long]() specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L) specificStartOffsets.put(new KafkaTopicPartition("myTopic", 1), 31L) specificStartOffsets.put(new KafkaTopicPartition("myTopic", 2), 43L) myConsumer.setStartFromSpecificOffsets(specificStartOffsets)

注意:当作业从故障中自动还原或使用checkpoint手动还原时,会从保存的状态中的offset继续消费,而不会再次使用这些设置。

kafka消费容错(checkpoint机制)

开启checkpoint机制后,flink会在消费kafka时,定期保存kafka的offset以及工作的状态(包括计算结果),且保证数据一致性。如果任务发生错误,flink会从checkpoint中恢复计算状态,并从保存的offset处开始重新消费数据。 

因此,保存checkpoint的间隔时间决定了当任务失败是丢失数据的多少。

checkpoint需要在调用addsink前,调用上下文环境对象的enableCheckpointing方法,参数为保存checkpoint的时间间隔,单位为毫秒。代码如下:

val env = StreamExecutionEnvironment.getExecutionEnvironment() env.enableCheckpointing(5000) // checkpoint every 5000 msecs 分区感知

flink消费kafka支持动态地感知kafka分区变化,当kafka新建分区时,flink可以发现这个新分区,并且可以精准一次地消费它们。在分区元数据初始检索之后发现的所有分区(即,当作业开始运行后发现的分区)将从最早的偏移量开始消费。

默认情况下,分区感知是禁用的,通过在properties 中设置 flink.partition-discovery.interval-millis 的参数来开启分区感知,参数为一个非负数的值,表示检查分区变化的间隔时间,单位是毫秒。

flink也可以基于主题名称使用正则表达式来匹配主题,如:

val env = StreamExecutionEnvironment.getExecutionEnvironment() val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") properties.setProperty("group.id", "test") val myConsumer = new FlinkKafkaConsumer08[String]( java.util.regex.Pattern.compile("test-topic-[0-9]"), new SimpleStringSchema, properties) val stream = env.addSource(myConsumer)

在上面的例子中,flink会消费所有以"test-topic-"开头,以数字结尾的所有topic。

offset提交配置

kafka会把flink消费到的offset提交到kafka内置的topic里(0.8版本的kafka保存到zookeeper),但是flink不会基于这些offset来作为容错机制

,kafka保存的offset仅仅作为kafka监控消费状态之用。

根据是否启用checkpoint,对offset提交也有不同的方式:

启用checkpoint:如果启用了checkpoint,那么flink会先将offset和state保存到checkpoint之后,再将offset提交给kafka。这样可以确保kafka和checkpoint中保存的offset是一致的。可以使用setCommitOffsetsOnCheckpoints(boolean)设置是否提交offset到kafka,默认为true。如果调用了setCommitOffsetsOnCheckpoints,则setCommitOffsetsOnCheckpoints提交的参数将会覆盖Properties中配置的参数。

禁用checkpoint:如果禁用了checkpoint,那么flink则依赖提交到kafka中的offset进行消费,可以在Properties中使用enable.auto.commit (0.8版本的kafka是auto.commit.enable) 或者 auto.commit.interval.ms 来设置是否自动提交offset以及提交间隔。

提取时间戳与生成watermark

kafka数据中可能带有事件时间戳,时间戳和watermark在另一篇文章中有详说,这里不再赘述。如果不需要使用事件时间戳,可以跳过本节。

设置watermark生成器以及注册时间戳戳是通过调用kafka消费者对象的assignTimestampsAndWatermarks方法,传入自定义的watermark生成器,如:

val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") // only required for Kafka 0.8 properties.setProperty("zookeeper.connect", "localhost:2181") properties.setProperty("group.id", "test") val myConsumer = new FlinkKafkaConsumer08[String]("topic", new SimpleStringSchema(), properties) myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter()) stream = env .addSource(myConsumer) .print()

watermark以及如何自定义watermark生成器(分配器),参考:https://blog.csdn.net/x950913/article/details/106246807

 

Flink之Kafka生产者

flink的kafka生产者类名为FlinkKafkaProducer011 (如果是Kafka 0.10.0.x 版本,则是FlinkKafkaProducer010;如果kafka版本高于1.0.0,则是FlinkKafkaProducer )。可以使用生产者对象将数据写入一个或多个topic。

代码示例:

val stream: DataStream[String] = ... val myProducer = new FlinkKafkaProducer011[String]( "localhost:9092", // broker list "my-topic", // target topic new SimpleStringSchema) // serialization schema // versions 0.10+ allow attaching the records' event timestamp when writing them to Kafka; // this method is not available for earlier Kafka versions myProducer.setWriteTimestampToKafka(true) stream.addSink(myProducer)

 

序列化器

参考kafka消费者的反序列化器。

kafka生产者分区器

如果没有设置分区器的话,flink会使用自带的FlinkFixedPartitioner 进行分区,默认是每个并行的子任务产生一个分区,即分区数等于并行度。

可以通过继承FlinkKafkaPartitioner类自定义分区器。所有版本都支持自定义分区器。

注意:分区器必须是可序列化的(serializable),因为它们会被发送到各个flink节点上。而且,分区器不会被保存到checkpoint,所以不要在分区器中保存状态,否则任务失败后,分区器里的状态都将丢失。

也可以不使用分区器,而依赖序列化器对每条数据指定其分区。如果要这样的话,那么必须在设置分区器时,使用null作为分区器(必须指定null,因为上面说了如果不指定分区器的话,会使用默认的FlinkFixedPartitioner 分区器)。

kafka生产者容错机制

kafka 0.8版本

0.8版本的kafka是不支持精准一次性和至少一次性容错的。

kafka 0.9和0.10版本

开启checkpoint之后,0.9和0.10版本的kafka支持至少一次消费。

除了开启checkpoint以外,还应该使用setLogFailuresOnly(boolean) 和setFlushOnCheckpoint(boolean) 方法配置其他参数。

setLogFailuresOnly : 默认为false。设置为true后,当发生异常时,仅记录错误信息,而不抛出异常。可以理解为,如果发生了异常,那么该条数据也被认为已经成功提交给kafka了。所以,如果想要保证至少一次提交的话,则把参数设置为false。setFlushOnCheckpoint : 默认为true。当开启时,flink在保存checkpoint时,会等待kafka返回ack确认之后,才保存offset和state。这可以确保在offset和state写入checkpoint之前,所有的数据都已经提交给kafka了。如果要保证至少一次提交的话,那么必须设置为true。

如果对接0.9与0.10版本的kafka想要保证至少一次提交,则必须开启checkpoint以及保证setLogFailuresOnly 为false,setFlushOnCheckpoint 为true。

注意:默认的提交重试次数为0,所以,当setLogFailuresOnly 为false的话,如果发生了错误,则该数据的提交立即失败。默认情况为0是为了防止生成重复数据,生产环境中建议提高重试次数。

kafka 0.11及更高版本

如果使用了checkpoint,那么FlinkKafkaProducer011 (如果kafka版本高于1.0.0,则是FlinkKafkaProducer )可以保证精准一次性。

开启checkpoint后,可以选择三种提交模式。通过对FlinkKafkaProducer011 设置semantic的参数选择:

Semantic.NONE:Flink不保证任何精准性,数据可能会丢失,也可能会被重复提交。Semantic.AT_LEAST_ONCE:默认配置,保证至少一次提交,数据不会丢失,但是可能会重复。与kafka 0.9与0.10版本中setFlushOnCheckpoint 为true的情况一样。Semantic.EXACTLY_ONCE: 保证精准一次性提交。使用kafka的事务机制实现(配合flink的二阶段提交,后面另开文章详说checkpoint机制及二阶段提交机制)。官方提示,如果写入kafka时使用了事务机制,那么需要修改 isolation.level  的设置(read_committed 或read_uncommitted 后者为默认值,应该修改为前者),原因在后面的“注意”说。

注意:Semantic.EXACTLY_ONCE模式下,发生故障后,需要从checkpoint中恢复state后才能继续提交。如果恢复时间(或者flink故障时间)大于kafka事务的超时时间,那么数据会丢失。即,第一次提交之后,flink挂掉,当flink恢复后,进行二阶段提交,但是事务已经超时了,那么这次提交的数据就丢失了。基于此,可以适当调整kafka的事务超时时间。

kafka默认超时时间由 transaction.max.timeout.ms 设置,默认15分钟。当二阶段提交的间隔大于15分钟,则该事务失败。FlinkKafkaProducer011将其默认值改为1小时。所以在使用精准一次性语义时,应该适当提高该值。

如果kafka的消费者使用了read_committed 模式,那么未完成的事务在提交之前,这个事务之后的所有数据都不会被消费者读取。举个例子,如果两个事务时间线如下:

发送事务A;发送事务B;提交事务B;

尽管事务B先于事务A提交,但是事务A仍然没有提交,所以消费者依然读取不到事务B发送的数据,除非使用的是read_uncommitted 模式。

再次注意: Semantic.EXACTLY_ONCE 模式下,每个 FlinkKafkaProducer011  实例都会使用一个固定大小的kafka生产者池,其中每个kafka生产者都有一个checkpoint。如果当前并发的checkpoint数量大于这个pool的大小,那么flink会报错,并导致应用失败。所以建议适当提高这个pool的大小。

最后注意:上面说到,当事务迟迟没有提交,消费者又处于read_committed模式下,那么这个事务之后提交的数据也无法被读取。那么就有这么一种情况,当flink程序发送了一个事务,但是没有提交,第一个checkpoint也还没有来得及生成之前,flink就挂了,当flink重启后,checkpoint中没有之前的信息,无法提交这个事务,也不知道存在这个未提交的事务,又继续发送其他数据,那么消费者是消费不到这些数据的,而且这些数据也有可能会被回滚(猜测,有待验证)。所以这是不安全的,要保证在生成checkpoint之前,flink程序不要挂掉。

这个参数FlinkKafkaProducer011.SAFE_SCALE_DOWN_FACTOR 似乎与这个机制有关,先放着,有时间再仔细看看。

/** * This coefficient determines what is the safe scale down factor. * *

If the Flink application previously failed before first checkpoint completed or we are starting new batch * of {@link FlinkKafkaProducer011} from scratch without clean shutdown of the previous one, * {@link FlinkKafkaProducer011} doesn't know what was the set of previously used Kafka's transactionalId's. In * that case, it will try to play safe and abort all of the possible transactionalIds from the range of: * {@code [0, getNumberOfParallelSubtasks() * kafkaProducersPoolSize * SAFE_SCALE_DOWN_FACTOR) } * *

The range of available to use transactional ids is: * {@code [0, getNumberOfParallelSubtasks() * kafkaProducersPoolSize) } * *

This means that if we decrease {@code getNumberOfParallelSubtasks()} by a factor larger than * {@code SAFE_SCALE_DOWN_FACTOR} we can have a left some lingering transaction. */ public static final int SAFE_SCALE_DOWN_FACTOR = 5; 使用kafka时间戳和flink事件时间

在kafka 0.10及以后的版本中,kafka的消息中支持携带时间戳。这个时间戳可以是事件时间,也可以是消息到kafka中的时间。引入事件时间参考https://blog.csdn.net/x950913/article/details/106246807。

如果在flink中设置时间语义为事件时间 TimeCharacteristic.EventTime,那么 FlinkKafkaConsumer010  会发出带有事件时间戳的数据。设置时间语义代码如下:

StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

如果要kafka数据携带到达kafka的时间戳,则无需在生产时定义,默认就是到达kafka的时间。

无论使用哪种时间戳,都需要使用flink-kafka链接器的配置对象调用setWriteTimestampToKafka为true。

FlinkKafkaProducer010.FlinkKafkaProducer010Configuration config = FlinkKafkaProducer010.writeToKafkaWithTimestamps(streamWithTimestamps, topic, new SimpleStringSchema(), standardProps); config.setWriteTimestampToKafka(true); 数据丢失

尽管设置精准一次性生产后,以下配置的默认设置也可能导致数据丢失,值得注意:

ackslog.flush.interval.messageslog.flush.interval.mslog.flush.*

 



【本文地址】


今日新闻


推荐新闻


    CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3